package com.hgvip.autoCommit;

import com.hgvip.common.KafkaParams;
import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by arnold.zhu on 2017/8/2.
 */
public class Procedure {

    private static Logger logger = LoggerFactory.getLogger(Procedure.class);

    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        // broker地址
        props.put("bootstrap.servers", KafkaParams.BOOTSTRAP_SERVERS);
        // retries，如果请求失败，生产者会自动重试，我们指定是0次，如果启用重试，则会有重复消息的可能性。
        props.put("retries", KafkaParams.RETRIES);
        // (生产者)缓存每个分区未发送消息。缓存的大小是通过 batch.size 配置指定的。
        props.put("batch.size", KafkaParams.BATCH_SIZE);
        // 默认缓冲可立即发送，即遍缓冲空间还没有满，但是，如果你想减少请求的数量，可以设置linger.ms大于0
        props.put("linger.ms", KafkaParams.LINGER_MS);
        // 将用户提供的key和value 转换成String类型
        props.put("key.serializer", KafkaParams.KEY_DESERIALIZER);
        props.put("value.serializer", KafkaParams.VALUE_DESERIALIZER);
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 500; i++) {
            // send()方法是异步的，添加消息到缓冲区等待发送，并立即返回。生产者将单个的消息批量在一起发送来提高效率。
            producer.send(new ProducerRecord<>("my-topic22", Integer.toString(i), Integer.toString(i)), (metadata, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                }
                logger.info("The offset of the record we just sent is: " + metadata.offset() + " Partition is :" + metadata.partition());
            });
        }
        producer.close();
    }

}
